/*
 * Copyright (c) 2018-2024 北京华恒盛世科技有限公司
 * QUICKPOOL is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *     http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
*/

#include <getopt.h>

#include "tclinfo.h"
#include "jm.h"

#define CHECK_JS_TIME 10

struct config_param jmParams[] = {
    {"STDOUT_DIRECT", BOOL_PARAM, -1, {FALSE}},
    {"JM_CONNTIMEOUT", INT_PARAM, -1, {0}},
    {"JM_RECVTIMEOUT", INT_PARAM, -1, {0}},
    {"GETPWNAM_RETRY", INT_PARAM, -1, {0}},
    {"SIGNAL_SIGSTOP", STRING_PARAM, -1, {NULL}},
    {"COMPUTE_NODE", BOOL_PARAM, -1, {FALSE}},
    {"IGNORE_CHECKSUM", BOOL_PARAM, -1, {FALSE}},
    {"SPECIFY_NCPUS", STRING_PARAM, -1, {NULL}},
    {"RENICE_NEVER_AT_RESTART", BOOL_PARAM, -1, {FALSE}},
    {NULL, STRING_PARAM, -1, {NULL}},
};

char *gstrEnvDir = NULL;
bool_t gboolJSReady = FALSE;
time_t gtimeNodeBoot = 0;
time_t gtimeJmStart = 0;
char gboolCheckFinishJobs = FALSE;
char gboolCheckFinishTasks = FALSE;
int gintJMSleepTime = DEFAULT_JM_SLEEPTIME;
int gintUpdateUsage = DEFAULT_RUSAGE_UPDATE_RATE;
int gintUpdateUsagePercent = DEFAULT_RUSAGE_UPDATE_PERCENT;
int gintJobTerminateInterval = DEFAULT_JOB_TERM_INTERVAL;
bool_t gboolStdoutDirect = FALSE;
uid_t guidManager = 0; // by default, root is manager.
int gintJsFd = -1;
int gintJmQuit = 0;
bool_t gboolJmRestart = 0;
bool_t gboolJmCheckConfig = FALSE;
int gintConfigError = 0;



struct loadIndex *gptrMyLoad;
char *gstrBossNode=NULL;
struct cluster *gptrMyCluster;
struct hostNode *gptrMyHost;
bool_t gboolMaster=FALSE;
bool_t gboolMasterCand=FALSE;
HASH_TABLE_T *ghashResource;

LIST_T *glistJmJobs;
LIST_T *glistJobTask;

static void check_master(time_t);
static char *jm_mastername(void);
static void start_master(time_t);
static int get_node_boottime(time_t *);

static void handle_channels(void);
static void jm_die_handler(int);
static void jm_child_handler(int);

static bool_t sboolJsStarted = FALSE;
static pid_t spidJs;
static short sshortJsExit = DAEMON_QUIT_INIT;
static time_t stimeCheckStartMaster;
static time_t stimeCheckJs;
static time_t stimeCheckWindow;
static int sintCheckMaster;

int
main(int argc, char **argv)
{
    save_daemon_path(argv[0]);

    bool_t fOpt=FALSE;
    while (TRUE) {
        int option_index = 0;
        static struct option long_options[] = {
            {"help", no_argument, 0, 'h'},
            {"version", no_argument, 0, 'V'},
            {"dir", required_argument, 0, 'd'},
            {"front", no_argument, 0, 'f'},
            {"checkconfig", no_argument, 0, 'C'},
            {NULL, 0, 0, 0}
        };

        int opt;
        opt = getopt_long(argc, argv, "hVd:fC", long_options, &option_index);
        if (-1 == opt) {
            break;
        }

        switch (opt) {
        case 'd':
            gstrEnvDir = optarg;
            gf_setenv("ENVDIR", gstrEnvDir);
            break;
        case 'f':
            fOpt = TRUE;
            break;
        case 'V':
            fputs(_QP_VERSION_, stdout);
            exit(0);
        case 'C':
            gf_setenv("RECONFIG_CHECK", (char *)"YES");
            gboolJmCheckConfig = TRUE;
            break;
        case 'h':
            fprintf(stdout, "jm usage: [-h] [-V] [-d env_dir] [-f]\n");
            exit(0);
        case '?':
        default:
            fprintf(stderr, "jm usage: [-h] [-V] [-d env_dir] [-f]\n");
            exit(1);
        }
    }

    struct config_param logParam[] = {
        {"LOGDIR", STRING_PARAM, -1, {NULL}},
        {NULL, STRING_PARAM, -1, {NULL}},
    };


    char myhostnm[MAXHOSTNAMELEN+1];
    gethostname(myhostnm, MAXHOSTNAMELEN);
    if (0 == strlen(myhostnm)){
        char *myhostname;
        if ((myhostname = qp_getmyhostname()) == NULL) {
            fprintf(stderr, "Failed to get my host name.\n");
            jm_die(DAEMON_QUIT_FATAL);
        }
    }

    init_config_param(logParam, gstrEnvDir);
    if (NULL != logParam[0].string_value) {

        char logfile[PATH_MAX+1];
        snprintf(logfile, PATH_MAX, "%s/jm.%s.log", logParam[0].string_value, myhostnm);
        open_log(logfile, "INFO", 0);
    }

    int i;
    if (!gboolJmCheckConfig){
        for (i=sysconf(_SC_OPEN_MAX); i>=3; i--) {
            close(i);
        }
    
    
        guidManager = getuid();
        if (0 != guidManager) {
            log_message(ERR, NOCLASS, "%s: real uid is %d, only root can start JM.\n",
                    argv[0], (int)getuid());
            exit(1);
        }

        if (geteuid() != 0) {
            log_message(ERR, NOCLASS, "%s: effective uid is %d, only root can start JM.\n",
                    argv[0], (int)geteuid());
            exit(1);
        }
    }

    umask(022);

    if (!fOpt && !gboolJmCheckConfig) {
        daemonize_myself();
    }

    if (init_config_param(jmParams, gstrEnvDir) < 0) {
        log_message(ERR, NOCLASS, "%s: init_config_param() failed: see above message.", __func__);
        jm_die(DAEMON_QUIT_CONFFILE);
    }

    if (gf_chan_init() < 0) {
        jm_die(DAEMON_QUIT_FATAL);
    }

    gf_set_hostfile(baseParams[BASE_CONFDIR].string_value);

    close_log();

    if (gboolJmCheckConfig) {
            gf_set_loglevel("WARNING");
            gf_set_logclass("NOCLASS");
            gf_set_timinglog(0);
    } else {
    
        char logfile[PATH_MAX+1];
        snprintf(logfile, PATH_MAX, "%s/jm.%s.log",
                 baseParams[BASE_LOGDIR].string_value, myhostnm);
        int cc = open_log(logfile, baseParams[BASE_LOG_MASK].string_value,
                      baseParams[BASE_JM_TIME_DEBUG].int_value);
        if (cc < 0) {
            fprintf(stderr, "failed to open log file[%s]: %s.\n", logfile, strerror(errno));
        }
    }

    gf_set_logclass(baseParams[BASE_JM_DEBUG].string_value);

    if (jmParams[JM_CONNTIMEOUT].int_value > 0)
        g_conntimeout = jmParams[JM_CONNTIMEOUT].int_value;

    if (jmParams[JM_RECVTIMEOUT].int_value > 0)
        g_recvtimeout = jmParams[JM_RECVTIMEOUT].int_value;

    gboolStdoutDirect = jmParams[STDOUT_DIRECT].bool_value;

    gtimeJmStart = time(NULL);

    init_nrm_info(gboolJmCheckConfig);
    
    if (gboolJmCheckConfig) {
        exit(gintConfigError);
    }

    if (0 > get_node_boottime(&gtimeNodeBoot)) {
        gtimeNodeBoot = 0;
        jm_die(DAEMON_QUIT_FATAL);
    }

    Signal_(SIGALRM, SIG_IGN);
    Signal_(SIGHUP,  SIG_IGN);
    Signal_(SIGTERM, (SIGFUNCTYPE) jm_die_handler);
    Signal_(SIGINT,  (SIGFUNCTYPE) jm_die_handler);
    Signal_(SIGCHLD, (SIGFUNCTYPE) jm_child_handler);
    Signal_(SIGPIPE, SIG_IGN);

    Signal_(SIGTTOU, SIG_IGN);
    Signal_(SIGTTIN, SIG_IGN);
    Signal_(SIGTSTP, SIG_IGN);

    glistJmJobs = gf_list_make("jm job list");
    if (NULL == glistJmJobs) {
        log_message(ERR, NOCLASS, "%s: gf_list_make(jm job list) failed", __func__);
        jm_die(DAEMON_QUIT_MEM);
    }

    glistJobTask = gf_list_make("jm job task list");
    if (NULL == glistJobTask) {
        log_message(ERR, NOCLASS, "%s: gf_list_make(jm job task list) failed", __func__);
        jm_die(DAEMON_QUIT_MEM);
    }

    if (NULL != gptrMyCluster && NULL != gptrMyCluster->boss_node) {

        FREEUP(gstrBossNode);
        gstrBossNode = copy_string(gptrMyCluster->boss_node->name);

        log_message(INFO, NOCLASS, "%s: cluster %s with master %s",
                    __func__, gptrMyCluster->name, gstrBossNode);
    }

    struct hostent *hostEnt;
    hostEnt = gf_gethostbyname(myhostnm);
    if (NULL == hostEnt) {
        log_message(ERR, NOCLASS, "%s: cannot get my host entry <%s>.",
                    __func__, myhostnm);
        jm_die(DAEMON_QUIT_FATAL);
    }

    struct sockaddr_in myAddr;
    memset(&myAddr, 0, sizeof(struct sockaddr_in));
    myAddr.sin_family = AF_INET;
    myAddr.sin_addr.s_addr = *(in_addr_t *)hostEnt->h_addr_list[0];
    myAddr.sin_port = htons(g_jm_port);

    if (gf_init_server(&myAddr) < 0) {
        log_message(ERR, NOCLASS, "gf_init_server failed.");
        jm_die(DAEMON_QUIT_FATAL);
    }

    if (chdir(gstrTmpDir) < 0) {
        log_message(ERR, NOCLASS, "%s: chdir(%s) failed due to %m", __func__, gstrTmpDir);
    }

    char *masterList = g_masterlist;
    unsigned int len = strlen(masterList);
    char *master = (char *)calloc(len+1, sizeof(char));
    if (NULL == master) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", len+1);
        jm_die(DAEMON_QUIT_MEM);
    }
    char *token;
    while (NULL != (token = get_string_token(&masterList, " ", master, len+1))) {

        len = strlen(masterList);
        if (strcmp(token, gptrMyHost->name) == 0) {

            gboolMasterCand = TRUE;
            if (get_lock()) {
                gboolMaster = TRUE;
                initialize_master();
                start_master(time(NULL));
            }

            break;
        }
    }
    FREEUP(master);

    if (!gboolMasterCand || (gboolMasterCand && !gboolMaster)) {
        gstrBossNode = jm_mastername();
    }
  
    if (!gboolMaster) {
        gboolJSReady = TRUE;
    }

    set_unrm_caller();

    for (i = gptrSharedConf->qpinfo->num_models; i < MAXMODELS; i++) {
        gptrSharedConf->qpinfo->cpu_factor[i] = 1.0;
    }

    log_message(INFO, NOCLASS, "%s: jm (re-)started", __func__);

    int fordev;
    if (getenv("DEVDEBUG") != NULL) {

        fordev = atoi(getenv("DEVDEBUG"));
        fordev = MIN(120, fordev);

        for (i=0; i<fordev;i++) {
            sleep(1);
        }
    }

    time_t timeUpdateLockFile=0;
    time_t timeJobChecking = 0;
    time_t timeLoadUpdate = 0;
    for(;;) {

        time_t timeSession = time(NULL);

        if (chdir(gstrTmpDir) < 0) {
            log_message(ERR, NOCLASS, "%s: chdir(%s) failed due to %m", __func__, gstrTmpDir);
        }

        if (!gboolMaster && NULL == gstrBossNode) {
            gstrBossNode = jm_mastername();
        }

        sigset_t oldsigmask, newmask;
        if (timeSession >= timeJobChecking + gintJMSleepTime) {

            sigemptyset(&newmask);
            sigaddset(&newmask, SIGCHLD);
            sigprocmask(SIG_BLOCK, &newmask, &oldsigmask);

            RECORD_FUNCTIME(0, check_jmjobs(timeSession), "check_jmjobs");
            RECORD_FUNCTIME(0, status_report(timeSession), "status_report");
            timeJobChecking = timeSession;

            RECORD_FUNCTIME(0, check_finish_tasks(), "check_finish_tasks");

            sigprocmask(SIG_SETMASK, &oldsigmask, NULL);

            if (!gboolMaster) {
                gf_handle_socket(0);
                handle_channels();
                gf_handle_socket(500);
            }
        }

        if (timeSession - timeUpdateLockFile >= gintJMSleepTime) {
            if (gboolMaster) {
                update_lock();
            }

            timeUpdateLockFile = timeSession;

            RECORD_FUNCTIME(0, check_master(timeSession), "check_master");
        }

        if (gboolCheckFinishJobs || gboolCheckFinishTasks) {

            sigemptyset(&newmask);
            sigaddset(&newmask, SIGCHLD);
            sigprocmask(SIG_BLOCK, &newmask, &oldsigmask);

            char buf[64+1];
            if (gboolCheckFinishJobs) {
                snprintf(buf, 64, "check_finish_jobs_%d", LIST_NUM_ENTS(glistJmJobs));
                RECORD_FUNCTIME(0, check_finish_jobs(), buf);
                gboolCheckFinishJobs = FALSE;
            }

            if (gboolCheckFinishTasks) {
                snprintf(buf, 46, "check_finish_tasks_%d", LIST_NUM_ENTS(glistJobTask));
                RECORD_FUNCTIME(0, check_finish_tasks(), buf);
                gboolCheckFinishTasks = FALSE;
            }

            sigprocmask(SIG_SETMASK, &oldsigmask, NULL);

            if (!gboolMaster) {
                gf_handle_socket(0);
                handle_channels();
                gf_handle_socket(500);
            }
        }

        if (timeSession - timeLoadUpdate >= (int)exchIntvl) {

            RECORD_FUNCTIME(0, getload_andsend(), "getload_andsend()");            

            if (timeSession - stimeCheckWindow > 60) {
                check_time_window();
                stimeCheckWindow = timeSession;
            }

            timeLoadUpdate = timeSession;
        }

        gf_handle_socket(0);
        handle_channels();
        gf_handle_socket(500);

        if (0 < gintJmQuit || gboolJmRestart) {
            break;
        }

        if (0 > gintJsFd && gboolJSReady) {
            notify_startup();
        }
    }

    if (gboolJmRestart) {
        relaunch();
    }

    jm_die(gintJmQuit);
    return 0;
} // end function main

/**
 * @brief       quit jm.
 *
 * @param[in]   sig          #1: killed signal
 *
 * @note        Let me log my quit reason before I quit.
 *              If I am master, need release lock file.
 */
void
jm_die(int sig)
{
    if (gboolMaster) {

        LIST_T *dynamicNodeList;
        dynamicNodeList = gf_list_make("JM dynamic node list");

        int ret;
        ret = load_dynamic_node(dynamicNodeList);
        if (1 == ret) {

            char strDynNodeFile[PATH_MAX+1];
            snprintf(strDynNodeFile, PATH_MAX, "%s/dynamic.nodes",
                     baseParams[BASE_WORKDIR].string_value);

            unlink(strDynNodeFile);

            LIST_ITERATOR_T iter;
            gf_listiterator_link(&iter, dynamicNodeList);

            struct dynamicNode *dynNode;
            for (dynNode = (struct dynamicNode *)gf_listiterator_curentry(&iter);
                 !gf_listiterator_isend(&iter);
                 gf_listiterator_getnext(&iter, (LIST_T **)&dynNode)) {

                if (NULL == dynNode || NULL == dynNode->node_req) {
                    continue;
                }

                log_addnode(dynNode->node_req);
            }
        }

        gf_list_free(dynamicNodeList, free_dynamic_node);

        release_lock(sig);
    }

    freeTclInterp();

    char myhost[MAXHOSTNAMELEN];
    if (gethostname(myhost, MAXHOSTNAMELEN) <0) {
        if (sig > 100) {
            log_message(ERR, NOCLASS, STR_FUNC_S_FAIL_M, __func__, "gethostname", myhost);
        }
        strcpy(myhost, "localhost");
    }

    if (sig > 0 && sig < 100) {
        log_message(ERR, NOCLASS, "Daemon on host <%s> received signal <%d>; exiting", myhost, sig);
    } else {
        switch (sig) {
        case DAEMON_QUIT_MEM:
            log_message(ERR, NOCLASS, "jm daemon on host <%s> failed in memory allocation; fatal error - exiting", myhost);
            break;
        case DAEMON_QUIT_CONFFILE:
            log_message(ERR, NOCLASS, "jm daemon cannot operate configuration file on host <%s>",
                        myhost);
            break;
        case DAEMON_QUIT_FATAL:
            log_message(ERR, NOCLASS, "jm daemon on host <%s> dying; fatal error - see above messages for reason", myhost);
            break;
        case JM_QUIT_LOCKFILE:
            log_message(ERR, NOCLASS, "jm daemon cannot operate lock file on host <%s>", myhost);
            break;
        case DAEMON_QUIT_RESTART:
            log_message(ERR, NOCLASS, "jm daemon on host <%s> restarting", myhost);
            break;
        case DAEMON_SHUTDOWN:
            log_message(ERR, NOCLASS, "jm daemon on host <%s> shutdown", myhost);
            break;
        default:
            log_message(ERR, NOCLASS, "jm daemon on host <%s> exiting, unknown exit code <%d>",
                        myhost, sig);
            break;
        }
    }

    exit(sig);
} // end function js_die

/**
 * @brief       restart myself.
 *
 * @note        Fork a child to start myself and parent die itself.
 */
void
relaunch(void)
{
    int pid;
    char *margv[6];
    int i = 0;

    pid = fork();

    if (0 > pid) {
        return;
    } else if (0 == pid) {
        sigset_t newmask;

        close_log();
        gf_child_chan_close(-1);

        gf_sleep(1000);

        char *strPathJM;

        char path[PATH_MAX+1];
        strPathJM = get_daemon_path("/jm", baseParams[BASE_SERVERDIR].string_value, path);
        if (NULL == strPathJM) {
            log_message(ERR, NOCLASS, "JM relaunch failed due to above reason.");
            exit(-1);
        }

        margv[0] = path;
        i = 1;
        if (NULL != gstrEnvDir) {
            margv[i] = (char *)"-d";
            i++;
            margv[i] = gstrEnvDir;
            i++;
        }
        margv[i] = NULL;
        sigemptyset(&newmask);
        sigprocmask(SIG_SETMASK, &newmask, NULL);

        execve(margv[0], margv, environ);
        log_message(ERR, NOCLASS, "JM relaunch failed: %m");
        exit(-1);
    }

    jm_die(DAEMON_QUIT_RESTART);
} // end function relaunch

/**
 * @brief       Handle channel.
 *
 * @note        Get socket fds which have data and handle them one by one.
 */
static void
handle_channels(void)
{
    int *handleFds=NULL;
    unsigned int numHandles = gf_get_fdlist(&handleFds);

    int jsState;
    jsState = gf_get_chan_state(gintJsFd);
    if (0 > jsState || CHANNEL_FREE == jsState) {
        gintJsFd = -1;
        if (gboolMaster) {
            gboolJSReady = FALSE;
        }
    }

    if (0 == numHandles) {
        FREEUP(handleFds);
        return;
    }

    sigset_t oldsigmask, newmask;
    sigemptyset(&newmask);
    sigaddset(&newmask, SIGCHLD);
    sigprocmask(SIG_BLOCK, &newmask, &oldsigmask);

    LIST_T **dataList;
    dataList = (LIST_T **)calloc(numHandles, sizeof(LIST_T *));
    if (NULL == dataList) {

        log_message(ERR, NOCLASS, "%s: calloc(%d) failed. %m", __func__, numHandles);
        FREEUP(handleFds);
        return;
    }

    unsigned int i;
    for (i=0; i<numHandles; i++) {

        int fd = handleFds[i];
        int numList = gf_get_data(handleFds[i], &(dataList[i]));

        if (0 >= numList) {
            continue;
        }

        socklen_t len;
        struct sockaddr_in from;
        len = sizeof(struct sockaddr_in);
        if (getpeername(fd, (struct sockaddr *)&from, &len) < 0) {
            log_message(ERR, NOCLASS, "%s: getpeername() failed on %d. %m", __func__, fd);

            gf_chan_close(fd);
            handleFds[i] = -1;
            continue;
        }

        char addrbuf[24+1];
        gf_sockaddr_str(&from, addrbuf);

        struct chan_data *recvData;
        struct chan_data *nextData;
        for (recvData = (struct chan_data *)dataList[i]->forw;
             recvData != (struct chan_data *)dataList[i]; recvData = nextData) {

            nextData = recvData->forw;

            if (NULL == recvData) {
                continue;
            }

            struct header *hdr = &(recvData->hdr);
            log_message(DEBUG, CHANNEL, "%s: got request[%d] on fd[%d] from <%s>.",
                        __func__, hdr->reqcode, handleFds[i], addrbuf);

            XDR xdrs;
            xdrmem_create(&xdrs, recvData->data, recvData->datalen, XDR_DECODE);

            char namebuf[MAXNAMELEN+1];
            if (gintJsFd == fd) { // Got startup reply from JS.

                if (gf_check_longconn(fd)) {
                    snprintf(namebuf, MAXNAMELEN, "handle_restart_reply_%s", addrbuf);
                    RECORD_FUNCTIME(0, handle_restart_reply(fd, hdr, &xdrs), namebuf);

                    gf_clean_recvdata(fd, recvData);
                    xdr_destroy(&xdrs);

                    continue;
                } else {
                    gintJsFd = -1;
                    if (gboolMaster) {
                        gboolJSReady = FALSE;
                    }
                }
            }

            struct header outhdr;
            char message[MAXMESSAGELENGTH+1];
            gf_init_header(&outhdr);

            struct auth auth;
            memset(&auth, 0, sizeof(struct auth));

            jm_request_t reqCode;
            reqCode = (jm_request_t)(hdr->reqcode);

            switch (reqCode) {
            case CMD_JM_RECONFIG:
            case CMD_JM_REBOOT:
            case CMD_JM_SHUTDOWN:
            case CMD_JM_DEBUG:
            case CMD_START_TASK:
            case JM_SIGNAL_TASK:
            case JM_GET_HOSTINFO:
            case JM_GET_RESOUINFO:
            case JM_LOAD_REQ:
            case JM_GET_INFO:
            case JM_GET_CLUSTERNAME:
            case JM_GET_MASTINFO:
            case JM_GET_CLUSINFO:
            case JM_ADD_HOST:
            case JM_RM_HOST:
                if (!gf_xdr_auth(&xdrs, (void *)&auth, hdr, NULL)) {
                    log_message(ERR, NOCLASS, "%s: failed to decode authentication info from %s.",
                                __func__, addrbuf);

                    snprintf(message, MAXMESSAGELENGTH,
                             "%s: failed to decode authentication info.", __func__);

                    outhdr.reqcode = ERROR_PERMISSION;
                    gf_errorback(fd, &outhdr, message);

                    gf_clean_recvdata(fd, recvData);
                    xdr_destroy(&xdrs);

                    continue;
                }

                break;
            default:
                break;
            }
        
            switch (reqCode) {
            case JS_NEW_JOB:
                snprintf(namebuf, MAXNAMELEN, "handle_newjob_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_newjob(fd, hdr, &xdrs), namebuf);
                break;
            case JS_SWIT_JOB:
                snprintf(namebuf, MAXNAMELEN, "handle_switchjob_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_switchjob(fd, hdr, &xdrs), namebuf);
                break;
            case JS_MODIFY_JOB:
                snprintf(namebuf, MAXNAMELEN, "handle_modifyjob_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_modifyjob (fd, hdr, &xdrs), namebuf);
                break;
            case JS_SIG_JOB:
                snprintf(namebuf, MAXNAMELEN, "handle_signaljob_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_signaljob(fd, hdr, &xdrs), namebuf);
                break;
            case JS_PROBE:
                snprintf(namebuf, MAXNAMELEN, "handle_probe_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_probe(fd, hdr, &xdrs), namebuf);
                break;
            case CMD_JM_REBOOT:
            case CMD_JM_RECONFIG:
                RECORD_FUNCTIME(0, handle_reboot(fd, hdr, &xdrs, &auth), "hanlde_reboot");
                break;
            case CMD_JM_SHUTDOWN:
                RECORD_FUNCTIME(0, handle_shutdown(fd, hdr, &xdrs, &auth), "handle_shutdown");
                break;
            case CMD_JM_DEBUG:
                RECORD_FUNCTIME(0, handle_jm_debug(fd, hdr, &xdrs, &auth), "handle_jm_debug");
                break;
            case JM_JOB_SETUP:
                snprintf(namebuf, MAXNAMELEN, "handle_job_setup_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_job_setup(fd, hdr, &xdrs), namebuf);
                break;
            case JM_GET_HOSTINFO:
                snprintf(namebuf, MAXNAMELEN, "handle_hostinfo_req_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_hostinfo_req(fd, hdr, &xdrs, &from), namebuf);
                break;
            case JM_GET_RESOUINFO:
                snprintf(namebuf, MAXNAMELEN, "handle_resourceinfo_req_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_resourceinfo_req(fd, hdr, &xdrs, &from), namebuf);
                break;
            case JM_LOAD_REQ:
                snprintf(namebuf, MAXNAMELEN, "handle_load_req_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_load_req(fd, hdr, &xdrs, &from), namebuf);
                break;
            case JM_GET_INFO:
                snprintf(namebuf, MAXNAMELEN, "handle_info_req_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_info_req(fd, hdr, &xdrs, &from), namebuf);
                break;
            case JS_READY:
                RECORD_FUNCTIME(0, handle_jsready(fd, hdr, &xdrs), "handle_jsready");
                break;
            case JM_GET_CLUSTERNAME:
                RECORD_FUNCTIME(2, handle_clustername_req(fd, hdr, &xdrs, &from),
                                "handle_clustername_info");
                break;
            case JM_GET_MASTINFO:
                RECORD_FUNCTIME(2, handle_masterinfo_req(fd, hdr, &xdrs, &from),
                                "handle_masterinfo_req");
                break;
            case JM_GET_CLUSINFO:
                handle_clusterinfo_req(fd, hdr, &xdrs, &from);
                break;
            case JM_LOAD_UPD:
                snprintf(namebuf, MAXNAMELEN, "handle_load_update_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_load_update(fd, hdr, &xdrs, &from), namebuf);
                break;
            case JM_STATIC_INFO:
                snprintf(namebuf, MAXNAMELEN, "handle_static_info_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_static_info(fd, hdr, &xdrs, &from), namebuf);
                break;
            case JM_ADD_HOST:
                handle_add_dynamic(fd, hdr, &xdrs, &from, &auth);
                break;
            case JM_RM_HOST:
                handle_delete_dynamic(fd, hdr, &xdrs, &from, &auth);
                break;
            case CMD_START_TASK:
                snprintf(namebuf, MAXNAMELEN, "handle_start_jobtask_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_start_jobtask(fd, hdr, &xdrs, &from), namebuf);
                break;
            case JM_SIGNAL_TASK:
                snprintf(namebuf, MAXNAMELEN, "handle_signal_jobtask_%s", addrbuf);
                RECORD_FUNCTIME(0, handle_signal_jobtask(fd, hdr, &xdrs, &from), namebuf);
                break;
            default:
            {
                outhdr.reqcode = ERROR_MESSAGE;

                snprintf(message, MAXMESSAGELENGTH, "%s: Unknown request type %d from host[%s].",
                         __func__, hdr->reqcode, addrbuf);
                gf_errorback(fd, &outhdr, message);

                break;
            }
            }

            gf_xdr_free(gf_xdr_auth, &auth);
            xdr_destroy(&xdrs);

            gf_clean_recvdata(fd, recvData);
        }
    }
    FREEUP(dataList);

    FREEUP(handleFds);
    sigprocmask(SIG_SETMASK, &oldsigmask, NULL);

    return;
} // end function handle_channels

/**
 * @brief       start Job Scheduler(JS).
 *
 * @param[in]   session      #1: session time
 *
 * @note        Start JS if it is not started.
 *              If JS was down for some reason, make sure not start
 *              it twice in 10 seconds.
 */
static void
start_master(time_t session)
{
    if (sboolJsStarted) {
        switch (sshortJsExit) {
        case DAEMON_QUIT_RECONFIG:
            break;
        case DAEMON_QUIT_LICENSE:
            // js quit due to license, do not start it again.
            return;
        case JS_QUIT_RESIGN:
        case DAEMON_QUIT_FATAL:
        case DAEMON_QUIT_MEM:
        default:
            if (session - stimeCheckStartMaster < CHECK_JS_TIME) {
                return;
            }
            break;
        }
    }

    stimeCheckStartMaster = session;

    char *strPathJS;
    char path[PATH_MAX+1];
    strPathJS = get_daemon_path("/js", baseParams[BASE_SERVERDIR].string_value, path);
    if (NULL == strPathJS) {
        log_message(ERR, NOCLASS, "%s: cannot start JS due to above reason.", __func__);
        return;
    }

    char *margv[6];
    margv[0] = path;

    int i=1;
    if (NULL != gstrEnvDir) {
        margv[i++] = (char *)"-d";
        margv[i++] = gstrEnvDir;
    }
    margv[i] = NULL;

    pid_t pid;
    pid = fork();
    if (0 > pid) {
        log_message(ERR, NOCLASS, "%s: fork() failed when starting js due to %m", __func__);
        return;
    }

    if (0 == pid) { // child, to start JS.

        close_log();
        gf_child_chan_close(-1);
        finalize_tcl();

        sigset_t newmask;
        sigemptyset(&newmask);
        sigprocmask(SIG_SETMASK, &newmask, NULL);

        execve(margv[0], margv, environ);
        log_message(ERR, NOCLASS, "%s: starting js, execve() failed: %m", __func__);

        exit(-1);
    }

    if (0 < pid) {
        spidJs = pid;
        log_message(INFO, NOCLASS, "%s: master js <%d> started by jm on host %s",
                    __func__, spidJs, gptrMyHost->name);
    }

    return;
} // end function start_master

/**
 * @brief       Get the time when node start up.
 *
 * @param[out]  boottime     #1: node start up time
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, syscall failure
 *
 * @note        Start JS if it is not started.
 *              If JS was down for some reason, make sure not start
 *              it twice in 10 seconds.
 */
static int
get_node_boottime(time_t *boottime)
{
    char *strEnvIFS=NULL;
    char *strOldIFS;
    strOldIFS = getenv("IFS");
    if (NULL != strOldIFS) {

        unsigned int len;
        len = strlen(strOldIFS) + strlen("IFS") + 2;
        strEnvIFS = (char *)calloc(len+1, sizeof(char));
        if (NULL == strEnvIFS) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", len+1);
    	    jm_die(DAEMON_QUIT_MEM);
        }

        snprintf(strEnvIFS, len, "IFS=%s", strOldIFS);
        putenv((char *)"IFS=");
    }

    FILE *fileUpTime;
    fileUpTime = popen("uptime", "r");
    if (NULL == fileUpTime) {

        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "popen");

        if (NULL != strEnvIFS) {
            putenv(strEnvIFS);
            FREEUP(strEnvIFS);
        }

        return -1;
    }

    char dummy[32], str1[32], str2[32], str3[32], str4[32];
    if (fscanf(fileUpTime, "%s %s %s %s %s %s",
               dummy, dummy, str1, str2, str3, str4) != 6) {

        log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "fscanf");
        pclose(fileUpTime);
        return -1;
    }

    int days;
    if (0 == strncmp(str2, "day", 3)) {
        days = atoi(str1);
        strcpy(str1, str3);
        strcpy(str2, str4);
    } else {
        days = 0;
    }

    int hour, minute;
    char *sp;
    sp = strchr(str1, ':');
    if (NULL != sp) {
        *sp = '\0';
        sp++;

        hour = atoi(str1);
        strcpy(str2, sp);
        sp = strchr(str2, ',');
        if (NULL != sp) {
            *sp = '\0';
        }
        minute = atoi(str2);
    } else {
        if (0 == strncmp(str2, "hr", 2)) {
            hour = atoi(str1);
            minute = 0;
        } else {
            hour = 0;
            if (0 == strncmp(str2, "min", 3)) {
                minute = atoi(str1);
            } else {
                minute = 0;
            }
        }
    }

    *boottime = time(NULL) - (days*ONEDAY + hour*ONEHOUR + minute*60);
    pclose(fileUpTime);

    if (NULL != strEnvIFS) {
        putenv(strEnvIFS);
        FREEUP(strEnvIFS);
    }

    return 0;
} // end function get_node_boottime

/**
 * @brief       check if Job Scheduler(JS) is alive or not.
 *
 * @param[in]   session      #1: session time
 *
 * @note        Do check for every 10 seconds.
 *              If JS is down, start a new one.
 *              If master is not active for long time, get lock file and 
 *              be a new master.
 */
static void
check_master(time_t session)
{
    if (session - stimeCheckJs < CHECK_JS_TIME) {
        return;
    }

    stimeCheckJs = session;

    if (NULL == gptrMyCluster || NULL == gptrMyCluster->boss_node) {
        sintCheckMaster ++;

        if (sintCheckMaster < 3 * gintMasterInactLimit) {
            return;
        }
    } else {

        sintCheckMaster = 0;
        FREEUP(gstrBossNode);
        gstrBossNode = copy_string(gptrMyCluster->boss_node->name);
    }

    if (NULL != gstrBossNode && issame_host(gstrBossNode, gptrMyHost->name)) {

        if (0 != spidJs) {

            if (kill(spidJs, 0) != 0) {
                start_master(session);
            } else {
                sboolJsStarted = TRUE;
            }
        } else {
            start_master(session);
        }
    } else if (gboolMasterCand && gptrMyCluster->number_inactivity >= gintMasterInactLimit) {
        // master has been down for a long time.
        if (get_lock()) {

            gf_chan_close(gintJsFd);
            gintJsFd = -1;
            initialize_master();
            start_master(session);
            gboolMaster = TRUE;
            gboolJSReady = FALSE;
        }
    } 

    return;
} // end function check_master

static void
jm_die_handler(int sig)
{
    gintJmQuit = sig;

    return;
}

/**
 * @brief       jm child handler.
 *
 * @note        child could be JS or job.
 */
static void
jm_child_handler(int sig)
{
    STATUS_WAIT_T status;
    struct rusage rusage;
    static short lastJsExitVal = DAEMON_QUIT_INIT;

    sigset_t newmask, oldmask;
    sigemptyset(&newmask);
    sigaddset(&newmask, SIGTERM);
    sigaddset(&newmask, SIGINT);
    sigaddset(&newmask, SIGCHLD);
    sigprocmask(SIG_BLOCK, &newmask, &oldmask);

    clean_usage(&rusage);

    pid_t pid;
    while ((pid = wait3(&status, WNOHANG, &rusage)) > 0) {
        if (pid == spidJs) { // JS exit
            int sig = WTERMSIG(status);

            sshortJsExit = WIFSIGNALED(status);
            if (0 < sshortJsExit) {
                if (WCOREDUMP(status)) {
                }
                sshortJsExit = sig;
                if (sshortJsExit != lastJsExitVal) {
                    lastJsExitVal = sshortJsExit;
                }
                continue;
            } else {
                sshortJsExit = WEXITSTATUS(status);

                if (sshortJsExit != lastJsExitVal) {
                    lastJsExitVal = sshortJsExit;
                }

                if (sshortJsExit == DAEMON_QUIT_RECONFIG) {
                    spidJs = 0;
                } else {
                    continue;
                }
            }

            continue;
        }

        // one of job exit.
        double ru_utime;
        ru_utime = rusage.ru_utime.tv_sec + rusage.ru_utime.tv_usec / 1000000.0;

        double ru_stime;
        ru_stime = rusage.ru_stime.tv_sec + rusage.ru_stime.tv_usec / 1000000.0;

        register float cpuTime;
        cpuTime = ru_utime + ru_stime;

        struct jobEntry *job;
        for (job=(struct jobEntry *)glistJmJobs->forw; job!=(struct jobEntry *)glistJmJobs;
             job=(struct jobEntry *)job->forw) {

            if (job->job_base.job_pid == pid) {

                job->execute_flag |= JOB_EXEC_FINISH;
                job->cpu_time = cpuTime;
                job->w_status = JOB_WSTATUS(status);
                job->comm_failcnt++;

                gboolCheckFinishJobs = TRUE;
                break;
            }
        }

        struct jobTask *task;
        for (task=(struct jobTask *)glistJobTask->forw; task!=(struct jobTask *)glistJobTask;
             task=(struct jobTask *)task->forw) {

            if (task->task_pid == pid) {

                task->end_time = time(NULL);
                task->w_status = JOB_WSTATUS(status);
                task->task_usage->utime = ru_utime;
                task->task_usage->stime = ru_stime;
                task->comm_failcnt++;

                gboolCheckFinishTasks = TRUE;
                break;
            }
        }
    }

    sigprocmask(SIG_SETMASK, &oldmask, NULL);

    return;
} // end function jm_child_handler

/**
 * @brief       Call master JM to get its name.
 *
 * @retval      not NULL     #1: succeed, master host name
 * @retval      NULL         #1: failed
 * 
 * @note        Let me know the master name when I'm starting.
 */
static char *
jm_mastername(void)
{
    struct auth auth;
    gf_set_auth(&auth, baseParams[BASE_AUTH].string_value);

    struct header hdr;
    gf_init_header(&hdr);
    hdr.reqcode = JM_GET_MASTINFO;

    unsigned int size = gf_xdrsize_auth(&auth) + HEADER_LEN;

    char *request;
    request = (char *)calloc(size, sizeof(char));
    if (NULL == request) {
        gf_xdr_free(gf_xdr_auth, (void *)&auth);
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", size);
        relaunch();
    }

    XDR xdrs;
    xdrmem_create(&xdrs, request, size, XDR_ENCODE);

    if (!gf_xdr_message(&xdrs, NULL, &hdr, &auth, NULL, NULL)) {
        log_message(ERR, NOCLASS, STR_FUNC_FAIL, __func__, "gf_xdr_message");
        gf_xdr_free(gf_xdr_auth, (void *)&auth);
        xdr_destroy(&xdrs);
        FREEUP(request);
        return NULL;
    }

    char *master=NULL;

    char *masterList = g_masterlist;
    unsigned int len = strlen(masterList);
    char *strNode = (char *)calloc(len+1, sizeof(char));
    if (NULL == strNode) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", len+1);
        xdr_destroy(&xdrs);
        FREEUP(request);
        relaunch();
    }
    char *token;
    while (NULL != (token = get_string_token(&masterList, " ", strNode, len+1))) {

        len = strlen(masterList);
        if (strcmp(token, gptrMyHost->name) == 0) {
            continue;
        }

        char *reply;
        struct header outhdr;
        int cc, fd = -1;
        cc = gf_callslave(token, g_jm_port, g_conntimeout, g_recvtimeout, 
                          XDR_GETPOS(&xdrs), request, &outhdr, &reply, &fd);

        if (fd > 0) {
            gf_chan_close(fd);
        }

        if (cc < 0) {
            continue;
        }

        quickpool_errno = outhdr.reqcode;
        if (quickpool_errno != 0 || reply == NULL) {
            continue;
        }

        XDR xdrs2;
        xdrmem_create(&xdrs2, reply, outhdr.length, XDR_DECODE);

        if (!gf_xdr_string(&xdrs2, &master, &outhdr, NULL)) {
            log_message(ERR, NOCLASS, STR_FUNC_FAIL, __func__, "gf_xdr_string");
            xdr_destroy(&xdrs2);
            FREEUP(reply);
            continue;
        }

        struct hostNode *node;
        node = (struct hostNode *)gf_hash_find(gptrMyCluster->node_table, master);
        if (NULL == node) {
            log_message(ERR, NOCLASS, "%s: node <%s is unknown to me, so cannot set it to master.",
                        __func__, gstrBossNode);
            relaunch();
        }

        gptrMyCluster->number_inactivity = 0;
        gptrMyCluster->boss_node = node;

        xdr_destroy(&xdrs2);
        FREEUP(reply);
        break;
    }
    FREEUP(strNode);

    gf_xdr_free(gf_xdr_auth, (void *)&auth);
    xdr_destroy(&xdrs);
    FREEUP(request);

    return master;
} // end function jm_mastername
